package com.kira.scaffoldmvc.Disruptor;

import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import com.lmax.disruptor.util.DaemonThreadFactory;
import java.util.concurrent.*;

public class DisruptorVsThreadPool {

    // 测试配置
    private static final int BUFFER_SIZE = 1024 * 8;  // Disruptor环形缓冲区大小
    private static final int NUM_EVENTS = 10_000_000; // 总事件数
    private static final int REPEAT = 5;              // 重复测试次数

    // Disruptor事件类
    static class LongEvent {
        long value;
        void set(long value) { this.value = value; }
    }

    public static void main(String[] args) throws Exception {
        System.out.println("开始性能测试...");
        System.out.println("事件数量: " + NUM_EVENTS);
        
        // 测试传统线程池
        long threadPoolTime = testThreadPool();
        System.out.printf("线程池执行耗时: %,d ms%n", threadPoolTime);
        
        // 测试Disruptor
        long disruptorTime = testDisruptor();
        System.out.printf("Disruptor执行耗时: %,d ms%n", disruptorTime);
        
        // 性能对比
        double ratio = (double) threadPoolTime / disruptorTime;
        System.out.printf("\nDisruptor性能是线程池的 %.2f 倍%n", ratio);
    }

    // 测试传统线程池(阻塞队列)
    private static long testThreadPool() throws InterruptedException, ExecutionException {
        long totalTime = 0;
        for (int i = 0; i < REPEAT; i++) {
            //线程池
            ExecutorService executor = Executors.newSingleThreadExecutor();
            //阻塞队列
            BlockingQueue<Long> queue = new ArrayBlockingQueue<>(BUFFER_SIZE);
            //并发控制器
            CountDownLatch latch = new CountDownLatch(1);
            
            // 消费者线程提交并启动,从阻塞队列里面获取数据并处理
            Future<Long> consumer = executor.submit(() -> {
                long start = System.currentTimeMillis();
                long count = 0;
                while (count < NUM_EVENTS) {
                    queue.take(); // 阻塞获取
                    count++;
                }
                long time = System.currentTimeMillis() - start;
                latch.countDown();
                return time;
            });
            
            // 生产者线程,往阻塞队列里面放入数据
            Thread producer = new Thread(() -> {
                try {
                    for (long j = 0; j < NUM_EVENTS; j++) {
                        queue.put(j); // 阻塞放入
                    }
                } catch (InterruptedException ignored) {}
            });
            //启动生产者线程
            producer.start();
            //等待任务执行结束
            latch.await();
            //计算耗时
            totalTime += consumer.get();

            //停止 消费者线程
            executor.shutdown();
            //打断 生产者线程
            producer.interrupt();
        }
        //返回blockingqueue结合异步线程消费总耗时
        return totalTime / REPEAT;
    }

    // 测试Disruptor
    private static long testDisruptor() throws InterruptedException {
        long totalTime = 0;
        for (int i = 0; i < REPEAT; i++) {
            // 1. 创建Disruptor
            Disruptor<LongEvent> disruptor = new Disruptor<>(
                LongEvent::new,
                BUFFER_SIZE,
                DaemonThreadFactory.INSTANCE,
                ProducerType.SINGLE,
                new YieldingWaitStrategy()
            );
            
            // 2. 设置事件处理器
            CountDownLatch latch = new CountDownLatch(1);
            final long[] startTime = new long[1];

            //disruptor事件处理器
            disruptor.handleEventsWith((event, sequence, endOfBatch) -> {
                if (event.value == 0) startTime[0] = System.currentTimeMillis();
                if (event.value == NUM_EVENTS - 1) latch.countDown();
            });
            
            // 3. 启动Disruptor
            RingBuffer<LongEvent> ringBuffer = disruptor.start();
            
            // 4. 发布事件-将事件提交到disruptor
            for (long j = 0; j < NUM_EVENTS; j++) {
                long sequence = ringBuffer.next();
                try {
                    //得到任务类
                    LongEvent longEvent = ringBuffer.get(sequence);
                    //将这个任务的ID赋值到对象里面
                    longEvent.set(j);
                } finally {
                    //发布任务-将队列放到里面由消费者进行处理
                    ringBuffer.publish(sequence);
                }
            }


            // 5. 等待处理完成并计时
            latch.await();
            //计算最终耗时
            long endTime = System.currentTimeMillis();
            totalTime += (endTime - startTime[0]);

            //关闭disruptor
            disruptor.shutdown();
        }
        //返回disruptor处理任务总耗时
        return totalTime / REPEAT;
    }
}